package tuyou.kafka.producer;

import java.util.Properties;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

/**
 * 消息生产者
 * 
 * @author：涂有
 * @date 2017年6月1日 上午11:27:47
 */
public class ProducerTest extends Thread {

	public static final String ip = "localhost:9092";
	private final KafkaProducer<String, String> producer;
	public static final String topic = "test";
	private final boolean isAsync = false;

	public ProducerTest() {
		Properties properties = new Properties();
		properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, ip);// broker群地址
//		properties.put(ProducerConfig.CLIENT_ID_CONFIG, "MsgProducer");// 自定义客户端id
		properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
				"org.apache.kafka.common.serialization.StringSerializer");// key序列化方式
		properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
				"org.apache.kafka.common.serialization.StringSerializer");// value序列化方式
		// properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, CustomPartitioner.class.getCanonicalName());//自定义分区函数

		// properties.load("properties配置文件");
		this.producer = new KafkaProducer<>(properties);
	}

	@Override
	public void run() {
		int msgNo = 0;

		while (true) {
			String msg = "Msg: " + msgNo;
			String key = msgNo++ + "";
			try {
				if (isAsync) {// 异步
					producer.send(
							new ProducerRecord<>(this.topic, key, msg),
							new MsgProducerCallback(System.currentTimeMillis(), key, msg));
				} else {// 同步
					producer.send(new ProducerRecord<>(this.topic, key, msg)).get();
				}

				System.out.println("发送：" + key + "-" + msg);
				Thread.sleep(1000);
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}

	/**
	 * 消息发送后的回调函数
	 */
	class MsgProducerCallback implements Callback {

		private final long startTime;
		private final String key;
		private final String msg;

		public MsgProducerCallback(long startTime, String key, String msg) {
			this.startTime = startTime;
			this.key = key;
			this.msg = msg;
		}

		public void onCompletion(RecordMetadata recordMetadata, Exception e) {
			long elapsedTime = System.currentTimeMillis() - startTime;
			if (recordMetadata != null) {
				System.out.println(msg + " be sended to partition no : " + recordMetadata.partition());
			}
		}
	}

	public static void main(String args[]) {
		new ProducerTest().start();// 开始发送消息
	}
}